package com.baishancloud.log.murloc

import com.baishancloud.log.common.util.TimeUtil
import com.baishancloud.log.format.scala.Origin
import com.baishancloud.log.murloc.tools.{IPData, IPInfo}
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

import scala.collection.mutable
import scala.util.matching.Regex

/**
 * 慢速ip计算
 *
 * @author ziqiang.wang 
 * @date 2021/12/23 16:32
 */
class BitRateCal(parameterTool: ParameterTool) extends ProcessFunction[Origin, Result] {

  //从url中提取码率
  val pattern: Regex = """.*_(\d+)[._]mp4.*""".r("rate")
  //输入参数，域名id
  val domainId: mutable.Map[String, Int] = mutable.Map[String, Int]()

  override def open(parameters: Configuration): Unit = {
    parameterTool.get(domainIds).split(";").foreach((x: String) => {
      val splits: Array[String] = x.split(":")
      domainId.put(splits(0), splits(1).toInt)
    })
  }

  override def processElement(value: Origin, ctx: ProcessFunction[Origin, Result]#Context, out: Collector[Result]): Unit = {
    pattern.findFirstMatchIn(value.url()).map(_.group("rate").toDouble).fold(return) { bitRate: Double => {
      // 如果日志中有码率，则开始计算
      val ipData: Option[IPData] = IPInfo.get(value.ip())
      ipData.fold(return) { ipData: IPData => {
        // 如果日志ip能找到对应IPData数据，则开始计算
        // 1 b/ms = 1 Kb/s
        val speed: Double = value.bodyBytesSent() * 8.0 / value.responseTime()
        val slow: Boolean = speed < bitRate
        Result(
          domainId(value.domain()),
          TimeUtil.timeAtString(value.datetimeOfUTC().toEpochSecond, 5 * 60),
          ipData.country,
          ipData.province,
          ipData.provider,
          if (slow) 1L else 0L,
          if (!slow) 1L else 0L
        )
      }
      }
    }
    }
  }
}
